package com.darrenchan.spark.sql

import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SparkSession}

/**
  * DataFrame和RDD的互操作二（通过编程的方式）可以事先不知道类型
  * 文件内容：
  * 1,zhangsan,20
  * 2,lisi,30
  * 3,wangwu,40
  */
object DataFrameRDDApp2 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local[2]").appName("DataFrameRDDApp2").getOrCreate()

    //主要业务逻辑
    //RDD ==> DataFrame
    val rdd = spark.sparkContext.textFile("infos.txt")

    val infoRDD = rdd.map(_.split(",")).map(line => Row(line(0).toInt, line(1), line(2).toInt))

    val schema = StructType(Array(StructField("id", IntegerType, true),
      StructField("name", StringType, true),
      StructField("age", IntegerType, true)))

    val infoDF = spark.createDataFrame(infoRDD, schema)
    infoDF.show()

    //1.spark sql查询方式一
    infoDF.filter(infoDF.col("age") > 20).show()

    //2.spark sql查询方式二
    infoDF.createOrReplaceTempView("infos") //设置表名
    spark.sql("select * from infos where age > 20").show()

    spark.sql(
        s"""
           |select wids, id, user_id
           |from bjh_dim_author_df
           |where event_day='20191023'
           |and array_contains(split(wids,'_'),'210')
         """.stripMargin).show(200)

    spark.sql(
      s"""
         |select wids, id, user_id
         |from bjh_dim_author_df
         |where event_day='20191024'
         |and array_contains(split(wids,'_'),'321')
         """.stripMargin).show(200)

    spark.sql(
        s"""
           |select wids, id, user_id
           |from bjh_dim_author_df
           |where event_day='20191023'
           |and source_type='muzhi'
         """.stripMargin).show(200)

    spark.sql(
        s"""
           |select *
           |from bjh_data_ods_thumb_doctor_article_2_di
           |where event_day='20191010'
         """.stripMargin)

    spark.stop()
  }

}
